07 FastAPI 연동 CRUD UI 만들기 backend | ✅ 편저: 코담 운영자

7강: FastAPI + SQLite + SQLAlchemy로 CRUD API 만들기

🔗 소스

---

🔧 SQLAlchemy란?

• SQLAlchemy는 Python의 대표적인 ORM (객체 관계 매핑) 라이브러리
• SQL 없이도 Python 객체를 통해 데이터베이스와 상호작용 가능
• 다양한 DB 지원: SQLite, PostgreSQL, MySQL 등
• 복잡한 쿼리, 관계 모델링까지 지원해 백엔드 API 구축에 필수적

---

🛠️ 환경 구성

1. 의존성 설치 (pip)

pip install fastapi sqlalchemy uvicorn pydantic

2. FastAPI 앱 초기화 + CORS 설정

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

• 개발 초기 단계에서는 모든 출처 허용
• 운영 환경에서는 보안상 allow_origins 제한 권장
• CORS 설정은 React, Vue 등 프론트엔드와 API 연동 시 필수

---

🗃️ SQLite + SQLAlchemy 설정

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

• SQLite는 파일 기반 DB이므로 check_same_thread=False 필수
• SessionLocal을 통해 요청마다 DB 세션을 생성
• declarative_base()로 모델 클래스의 기반 생성

---

👤 사용자 모델 정의

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

Base.metadata.create_all(bind=engine)

• SQLAlchemy 모델 클래스는 실제 DB 테이블과 매핑됨
• __tablename__은 DB 테이블명, 각 Column은 속성
• Base.metadata.create_all() 실행 시 테이블 생성

---

🔁 DB 세션 의존성

from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

• FastAPI에서는 Depends(get_db)를 통해 라우터마다 DB 세션 주입
• yield는 요청 완료 후 자동으로 세션 종료를 보장
• 커넥션 누수 방지 및 안정적 세션 관리

---

📦 Pydantic 모델 정의

from pydantic import BaseModel
from typing import Optional

class UserCreate(BaseModel):
    name: str
    email: str

class UserUpdate(BaseModel):
    name: Optional[str] = None
    email: Optional[str] = None

class UserResponse(BaseModel):
    id: int
    name: str
    email: str

    class Config:
        orm_mode = True

• 요청(Request): UserCreate, UserUpdate
• 응답(Response): UserResponse
• orm_mode = True → SQLAlchemy 객체를 자동으로 JSON 변환

---

📌 CRUD API 구현

✅ 사용자 목록 조회

@app.get("/users/", response_model=List[UserResponse])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return db.query(User).offset(skip).limit(limit).all()

• skip, limit: 페이지네이션 지원
• /users/?skip=0&limit=10 식으로 요청 가능

✅ 특정 사용자 조회

@app.get("/users/{user_id}", response_model=UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

• 없는 ID 요청 시 404 응답 반환

✅ 사용자 생성

@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

• db.commit()으로 DB 저장, refresh()로 ID 포함된 최신 상태 반환

✅ 사용자 수정

@app.put("/users/{user_id}", response_model=UserResponse)
def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")

    db_user.name = user.name if user.name else db_user.name
    db_user.email = user.email if user.email else db_user.email
    db.commit()
    db.refresh(db_user)
    return db_user

• 일부 필드만 수정 가능 (부분 업데이트)

✅ 사용자 삭제

@app.delete("/users/{user_id}", response_model=UserResponse)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")

    db.delete(db_user)
    db.commit()
    return db_user

• 삭제 성공 시 삭제된 사용자 정보를 반환

---

✅ 전체 코드

from typing import List, Optional

from fastapi import FastAPI, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

# FastAPI 인스턴스 생성
app = FastAPI()

# CORS 설정 (모든 출처 허용 - 운영 환경에서는 제한 필요)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 데이터베이스 연결 설정 (SQLite)
DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


# -----------------------------
# 모델 정의
# -----------------------------
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)


# 테이블 생성
Base.metadata.create_all(bind=engine)


# -----------------------------
# DB 세션 종속성
# -----------------------------
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# -----------------------------
# Pydantic 모델
# -----------------------------
class UserCreate(BaseModel):
    name: str
    email: str


class UserUpdate(BaseModel):
    name: Optional[str] = None
    email: Optional[str] = None


class UserResponse(BaseModel):
    id: int
    name: str
    email: str

    class Config:
        orm_mode = True


# -----------------------------
# API 라우터
# -----------------------------
@app.get("/users/", response_model=List[UserResponse])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return db.query(User).offset(skip).limit(limit).all()


@app.get("/users/{user_id}", response_model=UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user


@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


@app.put("/users/{user_id}", response_model=UserResponse)
def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")

    if user.name is not None:
        db_user.name = user.name
    if user.email is not None:
        db_user.email = user.email

    db.commit()
    db.refresh(db_user)
    return db_user


@app.delete("/users/{user_id}", response_model=UserResponse)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")

    db.delete(db_user)
    db.commit()
    return db_user

✅ 요약

• FastAPI + SQLAlchemy로 완전한 CRUD 기능 구축 완료
• Session, Base, Model, Pydantic, Depends()를 효율적으로 활용
• 실제 API 개발에 필요한 모든 구성 요소를 포괄
• 이후 강의(8강)에서는 프론트엔드 연동 및 관계형 데이터 설계를 이어서 학습합니다

📌 참고: 본 강의는 FastAPI 학습 시리즈 기반으로 제작되었습니다.